package com.ibm.datamodel;

import java.util.Collection;
import java.util.HashMap;
import java.util.Set;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonParser;
import com.ibm.cps.message.MessageFactory.MessageType;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ProcessorMetaMessage;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class DataSourceTopic extends ProcessorMetaMessage implements
		KryoSerializable {
	private String topicname;
	private String schemaid;

	public DataSourceTopic(String topicName, String schemaId) {
		this.topicname = topicName;
		this.schemaid = schemaId;
	}

	public DataSourceTopic(String tenantid, String sourceid, String topicName,
			String schemaId) {
		this.tenantid = tenantid;
		this.processorid = sourceid;
		this.topicname = topicName;
		this.schemaid = schemaId;
	}

	public DataSourceTopic(byte[] message) {
		String daString = new String(message);
		try {
			ObjectNode node = JSONUtility.fromObjectString(daString);
			tenantid = node.get(MessageFields.TENANTID).asText();
			processorid = node.get(MessageFields.PROCESSORID).asText();
			topicname = node.get(MessageFields.TOPICNAME).asText();
			schemaid = node.get(MessageFields.SCHEMAID).asText();
			type = MessageType.ADD_DATA_SOURCE_TOPIC;
		} catch (CPSException e) {
			e.printStackTrace();
		}

		// JsonParser parser = new JsonParser();
		// JsonObject dsObject = parser.parse(daString).getAsJsonObject();
		// tenantid = dsObject.get("tenantid").getAsString();
		// processorid = dsObject.get("processorid").getAsString();
		// topicname = dsObject.get("topicname").getAsString();
		// schemaid = dsObject.get("schemaid").getAsString();
		// type = dsObject.get("type").getAsString();
	}

	public String getTopicName() {
		return topicname;
	}

	public String getSchemaId() {
		return schemaid;
	}

	public String getUserid() {
		return tenantid;
	}

	public String getSourceid() {
		return processorid;
	}

	public String getJsonString() {
		Gson gson = new Gson();
		String jsonString = gson.toJson(this);
		ObjectNode objectNode;
		try {
			objectNode = JSONUtility.fromObjectString(jsonString);
			objectNode.remove(MessageFields.TYPE);
			objectNode.put(MessageFields.TYPE, "input");
			objectNode.remove("deletion");
			return objectNode.toString();
		} catch (CPSException e) {
			e.printStackTrace();
		}
		return null;
	}

	public static String getCollectionInputs(Collection<DataSourceTopic> topics) {
		JsonArray arrayNode = new JsonArray();
		JsonParser parser = new JsonParser();
		for (DataSourceTopic topic : topics) {
			arrayNode.add(parser.parse(topic.getJsonString()));
		}
		return arrayNode.toString();
	}

	@Override
	public void write(Kryo kryo, Output output) {
		output.writeString(tenantid);
		output.writeString(processorid);
		output.writeString(topicname);
		output.writeString(schemaid);
		kryo.writeObject(output, type);
	}

	@Override
	public void read(Kryo kryo, Input input) {
		tenantid = input.readString();
		processorid = input.readString();
		topicname = input.readString();
		schemaid = input.readString();
		type = kryo.readObject(input, MessageType.class);
	}

	@Override
	public void updateProcessorMap(
			HashMap<String, Set<String>> parent2Successors) {
		// nothing to do
	}
}
